package cn.monkey.io.transport.netty;

import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.ChannelGroupFuture;
import io.netty.channel.group.ChannelGroupFutureListener;
import reactor.core.publisher.Mono;

public class DefaultChannelGroupOutbound implements ChannelGroupOutbound {
    private final ChannelGroup channelGroup;

    public DefaultChannelGroupOutbound(ChannelGroup channelGroup) {
        this.channelGroup = channelGroup;
    }

    @Override
    public Mono<Void> close() {
        ChannelGroupFuture future = this.channelGroup.close();
        return Mono.create(sink -> future.addListener((ChannelGroupFutureListener) channelFutures -> {
            if (channelFutures.isSuccess()) {
                sink.success();
                return;
            }
            sink.error(channelFutures.cause());
        }));
    }

    @Override
    public Mono<Void> write(Object data) {
        ChannelGroupFuture groupFuture = this.channelGroup.write(data);
        return Mono.create(sink -> groupFuture.addListener((ChannelGroupFutureListener) channelFutures -> {
            if (channelFutures.isSuccess()) {
                sink.success();
                return;
            }
            sink.error(channelFutures.cause());
        }));
    }

    @Override
    public Mono<Void> writeAndFlush(Object data) {
        ChannelGroupFuture channelGroupFuture = this.channelGroup.writeAndFlush(data);
        return Mono.create(sink -> channelGroupFuture.addListener((ChannelGroupFutureListener) channelFutures -> {
            if (channelFutures.isSuccess()) {
                sink.success();
                return;
            }
            sink.error(channelFutures.cause());
        }));
    }

    @Override
    public Mono<Void> flush() {
        this.channelGroup.flush();
        return Mono.empty();
    }

    @Override
    public ChannelGroup channelGroup() {
        return this.channelGroup;
    }
}
